package com.atguigu.producer;

import com.atguigu.partitioner.MyPartitioner;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author code1997
 */
@Slf4j
public class MyPartitionProducer {


    public static void main(String[] args) {
        //1 创建配置文件
        Properties properties = new Properties();
        //2 填写配置信息（链接，kv序列化）
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop02:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        //配置自定义的分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("kafka-test", "code1997", "code1997+++" + i),
                    ((metadata, exception) -> {
                        //回调函数
                        if (exception == null) {
                            log.info("发送消息成功,topic:{},offset:{},partition:{}", metadata.offset(), metadata.offset(), metadata.partition());
                        } else {
                            //说明发送消息失败
                            log.error("发送消息发生异常,topic:{},offset:{},partition:{}", metadata.offset(), metadata.offset(), metadata.partition(), exception);
                        }
                    }));
        }
        producer.close();
    }
}
